package com.bsj.travel.device.consumer.rocketmq;

import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import com.bsj.travel.config.rocketmq.constant.ConsumerGroupConstant;
import com.bsj.travel.constant.BaseServerTopicConstants;
import com.bsj.travel.constant.MqttTopicPowerConstants;
import com.bsj.travel.def.common.DO.ProductDO;
import com.bsj.travel.def.common.DO.ProductTopicDO;
import com.bsj.travel.def.common.VO.device.DeviceInfoVO;
import com.bsj.travel.def.global.mq.Rp;
import com.bsj.travel.device.cache.TopicCache;
import com.bsj.travel.device.service.ProductTopicService;
import com.bsj.travel.util.ExceptionUtil;
import com.bsj.travel.util.ProductTopicUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.MessageModel;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQPushConsumerLifecycleListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;

import java.util.HashSet;
import java.util.List;
import java.util.Set;

/**
 * @author yinhao
 * @version 1.0
 * @description 设备-产品信息修改相关RocketMq监听
 * @date 2023/11/21
 */
@Slf4j
@Component
@RocketMQMessageListener(topic = BaseServerTopicConstants.BASE_SERVER_CHANGE_TOPIC,
        consumerGroup = ConsumerGroupConstant.TRAVEL_BASE_PRODUCT_GROUP,
        selectorExpression = BaseServerTopicConstants.BASE_CHANGE_PRODUCT_UPDATE_TAG
                + "||" + BaseServerTopicConstants.BASE_CHANGE_PRODUCT_DELETE_TAG
                + "||" + BaseServerTopicConstants.BASE_CHANGE_TOPIC_UPDATE_TAG
                + "||" + BaseServerTopicConstants.BASE_CHANGE_TOPIC_DELETE_TAG,
        messageModel = MessageModel.BROADCASTING)
public class BaseActionMessageRocketmqConsumer implements RocketMQListener<MessageExt>, RocketMQPushConsumerLifecycleListener {

    @Autowired
    private ProductTopicService productTopicService;

    @Async("travelAsync")
    @Override
    public void onMessage(MessageExt ext) {
        if (ext == null) {
            log.info("产品修改消息为空topic: {}", BaseServerTopicConstants.BASE_SERVER_CHANGE_TOPIC);
            return;
        }
        String message = new String(ext.getBody());
        log.info("Device-Product同步RocketMQ数据-->Received msg: {}", message);
        Rp rp = JSON.parseObject(message, Rp.class);
        String tag = rp.getTag();
        String productKey = rp.getProductKey();
        try {
            switch (tag) {
                // 新增产品tag
                case BaseServerTopicConstants.BASE_CHANGE_PRODUCT_UPDATE_TAG: {
                    Set<ProductTopicDO> productTopicSet = TopicCache.getInstance().get(productKey);
                    if (productTopicSet == null) {
                        //新加入产品对应公共主题信息
                        List<ProductTopicDO> topicList = productTopicService.queryTopicByProductId(0L);
                        if (topicList.isEmpty()) {
                            break;
                        }
                        Set<ProductTopicDO> newPubTopics = ProductTopicUtil.addTopic(productKey, topicList);
                        productTopicSet = new HashSet<>(newPubTopics);
                        TopicCache.getInstance().put(productKey, productTopicSet);
                    }
                }
                break;
                // 删除产品tag
                case BaseServerTopicConstants.BASE_CHANGE_PRODUCT_DELETE_TAG: {
                    TopicCache.getInstance().remove(productKey);
                }
                break;
                // 新增主题tag
                case BaseServerTopicConstants.BASE_CHANGE_TOPIC_UPDATE_TAG: {
                    Set<ProductTopicDO> productTopicSet = TopicCache.getInstance().get(productKey);
                    // 新加入主题信息
                    ProductTopicDO productTopicDO = JSONObject.parseObject(JSONObject.toJSONString(rp.getData()), ProductTopicDO.class);
                    productTopicSet.add(productTopicDO);
                    TopicCache.getInstance().put(productKey, productTopicSet);
                }
                break;
                // 删除主题tag
                case BaseServerTopicConstants.BASE_CHANGE_TOPIC_DELETE_TAG: {
                    Set<ProductTopicDO> productTopicSet = TopicCache.getInstance().get(productKey);
                    ProductTopicDO productTopicDO = JSONObject.parseObject(JSONObject.toJSONString(rp.getData()), ProductTopicDO.class);
                    ProductTopicDO removeTopic = null;
                    for (ProductTopicDO oldTopic : productTopicSet) {
                        if (oldTopic.getTopicId().equals(productTopicDO.getTopicId())) {
                            removeTopic = oldTopic;
                            break;
                        }
                    }
                    if (removeTopic != null) {
                        productTopicSet.remove(removeTopic);
                        TopicCache.getInstance().put(productKey, productTopicSet);
                    }
                }
                break;
                default:
                    log.info("未知产品变更tag: {}, message:{}", tag, message);
                    break;
            }
        } catch (Exception e) {
            log.error("Device-Product消费产品RocketMQ数据-->消费失败，失败原因: {}", ExceptionUtil.getStackStr(e));
        }
    }

    @Override
    public void prepareStart(DefaultMQPushConsumer defaultMQPushConsumer) {
        // 设置消费者消息重试次数
        defaultMQPushConsumer.setMaxReconsumeTimes(3);
        //设置实例名称
        defaultMQPushConsumer.setInstanceName(MqttTopicPowerConstants.BASE_ACTION_INSTANCE_NAME);
    }
}
